/*
 * MIT License
 *
 * Copyright (c) 2024-2048 冰羽
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

package cn.star.framework.es.service;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;
import cn.star.framework.core.api.Pageable;
import cn.star.framework.es.core.EsQueryBuilder;
import cn.star.framework.es.core.Where;
import cn.star.framework.es.core.WhereGroup;
import cn.star.framework.es.entity.IndexWrapper;
import cn.star.framework.type.BooleanOperator;
import cn.star.framework.type.OrderType;
import cn.star.framework.type.QueryType;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.SearchScrollHits;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.FetchSourceFilter;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;

/**
 * AbstractService<br>
 *
 * @author zhaoweiping
 *     <p style='color: red'>Created on 2024-09-23 11:06:00
 * @since 3.0.0
 */
public class AbstractService implements IService {
  private final ElasticsearchRestTemplate template;

  public AbstractService(ElasticsearchRestTemplate template) {
    this.template = template;
  }

  @Override
  public IndexWrapper save(IndexWrapper wrapper) {
    List<Map<String, Object>> data = wrapper.getData();
    Assert.isFalse(CollectionUtil.isEmpty(data), "data is empty, save not allowed");

    return this.modify(wrapper, false);
  }

  @Override
  public IndexWrapper delete(IndexWrapper wrapper) {
    List<Map<String, Object>> data = wrapper.getData();
    Assert.isFalse(CollectionUtil.isEmpty(data), "primary keys is empty, delete not allowed");

    return this.modify(wrapper, true);
  }

  @Override
  public IndexWrapper update(IndexWrapper wrapper) {
    List<Map<String, Object>> data = wrapper.getData();
    Assert.isFalse(CollectionUtil.isEmpty(data), "data is empty, update not allowed");

    return this.modify(wrapper, false);
  }

  @Override
  public Map<String, Object> findById(IndexWrapper wrapper, String id) {
    EsQueryBuilder esQueryBuilder = wrapper.getQueryBuilder();

    QueryBuilder queryBuilder = QueryBuilders.termQuery(IndexWrapper.PRIMARY_KEY, id);
    NativeSearchQueryBuilder nativeSearchQueryBuilder =
        new NativeSearchQueryBuilder().withQuery(queryBuilder);

    if (esQueryBuilder != null && CollectionUtil.isNotEmpty(esQueryBuilder.getQueryColumns())) {
      nativeSearchQueryBuilder.withSourceFilter(
          new FetchSourceFilter(
              ArrayUtil.toArray(esQueryBuilder.getQueryColumns(), String.class), null));
    }

    NativeSearchQuery query = nativeSearchQueryBuilder.build();

    SearchHit<Map> hit =
        template.searchOne(query, Map.class, IndexCoordinates.of(wrapper.getIndexName()));

    return this.wrapper(hit);
  }

  @Override
  public List<Map<String, Object>> find(IndexWrapper wrapper) {
    long scrollTimeInMillis = 60000L;
    Set<String> scrollIds = CollectionUtil.newHashSet();
    List<Map<String, Object>> data = new ArrayList<>();

    NativeSearchQueryBuilder nativeSearchQueryBuilder = this.builder(wrapper);
    NativeSearchQuery query = nativeSearchQueryBuilder.build();

    SearchScrollHits<Map> response =
        template.searchScrollStart(
            scrollTimeInMillis, query, Map.class, IndexCoordinates.of(wrapper.getIndexName()));

    String scrollId = response.getScrollId();
    if (!response.hasSearchHits() && StrUtil.isNotBlank(scrollId)) {
      scrollIds.add(scrollId);
    }

    while (response.hasSearchHits()) {
      data.addAll(this.wrapper(response));
      if (StrUtil.isNotBlank(scrollId)) {
        response =
            template.searchScrollContinue(
                scrollId,
                scrollTimeInMillis,
                Map.class,
                IndexCoordinates.of(wrapper.getIndexName()));
        scrollIds.add(scrollId);
        scrollId = response.getScrollId();
      }
    }

    // 释放es端资源
    if (CollectionUtil.isNotEmpty(scrollIds)) {
      template.searchScrollClear(CollectionUtil.newArrayList(scrollIds));
    }

    return data;
  }

  @Override
  public Pageable<Map<String, Object>> pageable(IndexWrapper wrapper, int pageNo, int pageSize) {
    NativeSearchQueryBuilder nativeSearchQueryBuilder = this.builder(wrapper, pageNo - 1, pageSize);
    NativeSearchQuery query = nativeSearchQueryBuilder.build();
    SearchHits<Map> response =
        template.search(query, Map.class, IndexCoordinates.of(wrapper.getIndexName()));

    List<Map<String, Object>> data = this.wrapper(response);

    Pageable<Map<String, Object>> pageable = new Pageable<>();
    pageable.setPageNo(pageNo);
    pageable.setPageSize(pageSize);
    pageable.setData(data);
    pageable.setTotal(response.getTotalHits());

    return pageable;
  }

  // ------------------- private

  private String getPrimaryKeyValue(Map<String, Object> data) {
    Assert.isFalse(CollectionUtil.isEmpty(data), "data is empty, illegal operation");
    // 在存储es数据时不能包含_id字段为元数据字段
    // ElasticsearchException[Elasticsearch exception [type=mapper_parsing_exception, reason=failed
    // to parse field [_id] of type [_id] in document with id 'a575e1bf8dd3f3da0494aeb5249f7cb7'.
    // Preview of field's value: 'a575e1bf8dd3f3da0494aeb5249f7cb7']]; nested:
    // ElasticsearchException[Elasticsearch exception [type=mapper_parsing_exception, reason=Field
    // [_id] is a metadata field and cannot be added inside a document. Use the index API request
    // parameters.]];
    Object id = data.remove(IndexWrapper.PRIMARY_KEY);
    return id == null ? null : String.valueOf(id);
  }

  private List<BulkRequest> convert(
      String index, List<Map<String, Object>> data, int batchSize, boolean delete) {
    int size = data.size();
    List<BulkRequest> list = new ArrayList<>();

    for (int i = 0; i < size; i += batchSize) {
      List<Map<String, Object>> batch = data.subList(i, Math.min(i + batchSize, size));
      if (CollectionUtil.isNotEmpty(batch)) {
        BulkRequest bulk = new BulkRequest();
        for (Map<String, Object> document : batch) {
          String id = this.getPrimaryKeyValue(document);
          if (delete) {
            Assert.isFalse(StrUtil.isBlank(id), "id is empty, illegal delete");
            bulk.add(new DeleteRequest(index, id));
          } else if (StrUtil.isNotBlank(id)) {
            bulk.add(new UpdateRequest(index, id).doc(document).upsert(document));
          } else {
            bulk.add(new IndexRequest(index).source(document));
          }
        }

        list.add(bulk);
      }
    }

    return list;
  }

  private IndexWrapper modify(IndexWrapper x, boolean delete) {
    List<BulkRequest> bulks = this.convert(x.getIndexName(), x.getData(), x.getBatchSize(), delete);
    List<BulkResponse> responses = new ArrayList<>();
    for (BulkRequest bulk : bulks) {
      responses.add(template.execute(client -> client.bulk(bulk, RequestOptions.DEFAULT)));
    }
    x.setResult(responses);

    // 解析数据存储是否成功，若存在失败则直接抛出异常
    for (BulkResponse response : responses) {
      if (response.hasFailures()) {
        BulkItemResponse[] bulkItemResponses = response.getItems();
        if (ArrayUtil.isNotEmpty(bulkItemResponses)) {
          for (BulkItemResponse bulkItemResponse : bulkItemResponses) {
            if (bulkItemResponse.isFailed()) {
              throw new RuntimeException(bulkItemResponse.getFailure().getCause());
            }
          }
        }
      }
    }

    return x;
  }

  @SuppressWarnings({"unchecked", "rawtypes"})
  private Map<String, Object> wrapper(SearchHit<Map> hit) {
    Map map = hit.getContent();
    Map<String, Object> data = new HashMap<>();
    data.put(IndexWrapper.PRIMARY_KEY, hit.getId());
    if (CollectionUtil.isNotEmpty(map)) {
      Set<String> keys = map.keySet();
      for (String key : keys) {
        Object value = map.get(key);
        if (key != null && value != null) {
          data.put(key, value);
        }
      }
    }

    return data;
  }

  private List<Map<String, Object>> wrapper(SearchHits<Map> response) {
    return response.getSearchHits().stream().map(this::wrapper).collect(Collectors.toList());
  }

  private NativeSearchQueryBuilder builder(IndexWrapper wrapper, int pageNo, int pageSize) {
    NativeSearchQueryBuilder nativeSearchQueryBuilder = this.builder(wrapper);
    // 分页 从0开始 -1
    nativeSearchQueryBuilder.withPageable(PageRequest.of(pageNo, pageSize));

    return nativeSearchQueryBuilder;
  }

  private NativeSearchQueryBuilder builder(IndexWrapper wrapper) {
    EsQueryBuilder esQueryBuilder = wrapper.getQueryBuilder();
    NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
    if (esQueryBuilder != null) {
      // 条件
      nativeSearchQueryBuilder.withQuery(this.builder(esQueryBuilder.getWhereGroups()));
      // 排序
      if (CollectionUtil.isNotEmpty(esQueryBuilder.getSorters())) {
        List<String> asc = new ArrayList<>(), desc = new ArrayList<>();
        esQueryBuilder
            .getSorters()
            .forEach(
                i -> {
                  List<String> names = i.getNames();
                  if (CollectionUtil.isNotEmpty(names)) {
                    if (i.getType() == OrderType.DESC) {
                      desc.addAll(names);
                    } else {
                      asc.addAll(names);
                    }
                  }
                });
        if (!asc.isEmpty()) {
          nativeSearchQueryBuilder.withSort(
              Sort.by(ArrayUtil.toArray(asc, String.class)).ascending());
        }
        if (!desc.isEmpty()) {
          nativeSearchQueryBuilder.withSort(
              Sort.by(ArrayUtil.toArray(desc, String.class)).descending());
        }
      }
      // 字段 查询返回的列
      if (CollectionUtil.isNotEmpty(esQueryBuilder.getQueryColumns())) {
        nativeSearchQueryBuilder.withSourceFilter(
            new FetchSourceFilter(
                ArrayUtil.toArray(esQueryBuilder.getQueryColumns(), String.class), null));
      }
    }

    return nativeSearchQueryBuilder;
  }

  private QueryBuilder builder(List<WhereGroup> whereGroups) {
    BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

    if (CollectionUtil.isNotEmpty(whereGroups)) {
      for (WhereGroup whereGroup : whereGroups) {
        BooleanOperator operator = whereGroup.getOperator();
        List<QueryBuilder> queryBuilders =
            whereGroup.getWheres().stream().map(this::builder).collect(Collectors.toList());
        if (CollectionUtil.isNotEmpty(queryBuilders)) {
          BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
          queryBuilders.forEach(queryBuilder::must);
          if (operator == BooleanOperator.OR) {
            boolQueryBuilder.should(queryBuilder);
          } else {
            boolQueryBuilder.must(queryBuilder);
          }
        }
      }
    }

    return boolQueryBuilder;
  }

  private QueryBuilder builder(Where where) {
    QueryType type = where.getType();
    String name = where.getName();
    Object value = where.getValue();
    if (type == QueryType.EQ) {
      return QueryBuilders.termQuery(name, value);
    } else if (type == QueryType.NE) {
      return QueryBuilders.boolQuery().mustNot(QueryBuilders.termQuery(name, value));
    } else if (type == QueryType.GT) {
      return QueryBuilders.rangeQuery(name).gt(value);
    } else if (type == QueryType.GE) {
      return QueryBuilders.rangeQuery(name).gte(value);
    } else if (type == QueryType.LT) {
      return QueryBuilders.rangeQuery(name).lt(value);
    } else if (type == QueryType.LE) {
      return QueryBuilders.rangeQuery(name).lte(value);
    } else if (type == QueryType.BETWEEN) {
      return QueryBuilders.rangeQuery(name).gte(1).lte(1);
    } else if (type == QueryType.NOT_BETWEEN) {
      return QueryBuilders.boolQuery().mustNot(QueryBuilders.rangeQuery(name).gte(1).lte(1));
    } else if (type == QueryType.LIKE) {
      return QueryBuilders.wildcardQuery(name, "*" + value + "*");
    } else if (type == QueryType.NOT_LIKE) {
      return QueryBuilders.boolQuery()
          .mustNot(QueryBuilders.wildcardQuery(name, "*" + value + "*"));
    } else if (type == QueryType.LIKE_LEFT) {
      return QueryBuilders.wildcardQuery(name, value + "*");
    } else if (type == QueryType.LIKE_RIGHT) {
      return QueryBuilders.wildcardQuery(name, "*" + value);
    } else if (type == QueryType.NULL) {
      return QueryBuilders.boolQuery().mustNot(QueryBuilders.existsQuery(name));
    } else if (type == QueryType.NOT_NULL) {
      return QueryBuilders.existsQuery(name);
    } else if (type == QueryType.IN) {
      return QueryBuilders.termsQuery(name, value);
    } else if (type == QueryType.NOT_IN) {
      return QueryBuilders.boolQuery().mustNot(QueryBuilders.termsQuery(name, value));
    } else {
      throw new RuntimeException("Not support type: " + type);
    }
  }
}
